package com.seal.commons.infrastructure.rocketmq.sender;

import com.seal.commons.exception.MessageSendFailException;
import com.seal.commons.exception.MessageSendStatusUnknownException;
import com.seal.commons.exception.errors.MessageErrorCode;
import com.seal.commons.infrastructure.facade.jms.MessageCallback;
import com.seal.commons.infrastructure.facade.jms.MessageSender;
import com.seal.commons.utils.serializer.ObjectSerializer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;

/**
 * ClassName:RocketMqSender <br/>
 * Function: rocketMQ 消息发送. <br/>
 * Date:     2019-06-05 10:11 <br/>
 *
 * @author yong.huang
 * @see
 * @since JDK 1.8
 */
@Slf4j
public class RocketMQSender implements MessageSender {

    private DefaultMQProducer rocketMqProducer;

    public RocketMQSender(DefaultMQProducer producer){
        this.rocketMqProducer = producer;
    }


    @Override
    public void send(String topic, String tags, Object message) {
        Message msg = buildMessage(topic, tags, message);
        send(msg);
    }

    @Override
    public void sendAsync(String topic, String tags, Object message, MessageCallback callback) {
        Message msg = buildMessage(topic, tags, message);
        try {
            rocketMqProducer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    if(SendStatus.SEND_OK.equals(sendResult.getSendStatus())){
                        callback.onSuccess(sendResult.getMsgId(), sendResult.getMessageQueue().getBrokerName(),
                                sendResult.getMessageQueue().getQueueId());
                    }else{
                        log.warn("topic: {} tags: {} send status : {}", topic, tags, sendResult.getSendStatus());
                        callback.onUnknown(sendResult.getMsgId(), sendResult.getMessageQueue().getBrokerName(),
                                sendResult.getMessageQueue().getQueueId());
                    }
                }
                @Override
                public void onException(Throwable e) {
                    log.error("topic: {} tags: {} send error", topic, tags, e);
                    callback.onException(e);
                }
            });
        }catch (Exception e){
            callback.onException(e);
        }
    }


    @Override
    public void sendOnce(String topic, String tags, Object message) {
        Message msg = buildMessage(topic, tags,message);
        try {
            rocketMqProducer.sendOneway(msg);
        } catch (Exception e) {
            log.error(" topic:[{}] ,tags:[{}] sendOnce error", topic, tags);
            throw new MessageSendFailException(MessageErrorCode.SEND_FAILED);
        }
    }

    @Override
    public void sendDeplay(String topic, String tags, Object message, int deplayLevel) {
        Message msg = buildMessage(topic, tags, message);
        msg.setDelayTimeLevel(deplayLevel);
        send(msg);
    }

    private void send(Message msg){
        try {
            SendResult sendResult = rocketMqProducer.send(msg);
            log.info("topic:[{}] ,tags:[{}], sendResult:[{}]", msg.getTopic(), msg.getTags(), sendResult.getSendStatus());
            if(!SendStatus.SEND_OK.equals(sendResult.getSendStatus())){
                throw new MessageSendStatusUnknownException(MessageErrorCode.SEND_STATUS_UNKNOWN);
            }
        } catch (Exception e) {
            log.error("message send error , topic:[{}], tags:[{}]", msg.getTopic(), msg.getTags(), e);
            throw new MessageSendFailException(MessageErrorCode.SEND_FAILED, e);
        }
    }


    private Message buildMessage(String topic, String tags, Object messgae){
        Message msg = new Message();// body
        msg.setTopic(topic);
        msg.setTags(tags);
        byte[] bytes = ObjectSerializer.serialize(messgae);
        msg.setBody(bytes);
        return msg;
    }
}
